package temp.casia.isiteam.zrb.hbase.client;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import temp.casia.isiteam.zrb.hbase.client.bean.HBGovContacts;
import temp.casia.isiteam.zrb.hbase.client.bean.HBGovernment;
import temp.casia.isiteam.zrb.hbase.client.bean.HbaseStatus;
import temp.casia.isiteam.zrb.hbase.client.bean.table.HbaseTableGovernment;
import temp.casia.isiteam.zrb.hbase.client.util.Commons;
import temp.casia.isiteam.zrb.hbase.client.util.MD5Util;

/**
 * 政府招商部门信息操作类<br>
 * 提供HBase中政府相关表的查询、写入
 * @author 彭鑫
 * @createDate 2015年6月29日
 */
public class GovHbaseClient {
	protected Configuration config;
	private Logger logger = Logger.getLogger(GovHbaseClient.class);
	/**
	 * 表名
	 */
	protected Map<String, TableName> tableNameMap;
	protected String quorum;

	/**
	 * 使用默认的master地址
	 */
	public GovHbaseClient(){
		this.quorum = Commons.quorum;
		this.init();
	}
	
	/**
	 * 指定master地址
	 * @param quorum
	 */
	public GovHbaseClient(String quorum){
		this.quorum = quorum;
		this.init();
	}
	
	protected void init() {
		Connection conn = null;
		try {
			config = HBaseConfiguration.create();
			config.set("hbase.zookeeper.quorum", this.quorum);
			config.set("hbase.client.retries.number", "3");
			config.set("zookeeper.recovery.retry", "3");
			
			conn = ConnectionFactory.createConnection(config);
			Admin admin = conn.getAdmin();
			TableName[] tableAry = admin.listTableNames("project_government");
			this.tableNameMap = new HashMap<String, TableName>();
			for (TableName tableName : tableAry) {
				this.tableNameMap.put(tableName.getNameAsString(), tableName);
			}
		} catch (Exception e) {
			this.logger.error("", e);
		} finally {
			this.close(conn, null);
		}
	}
	
	/**
	 * 查询招商单位信息
	 * @param provincePostcode
	 * @param cityPostcode
	 * @param countyPostcode
	 * @param govName
	 * @return
	 */
	public HBGovernment getgovInfo(String provincePostcode, String cityPostcode, String countyPostcode, String govName){
		String rowKey = provincePostcode + "_" + cityPostcode + "_" + countyPostcode + "_" + MD5Util.createMd5(govName);
		return this.getgovInfo(rowKey);
	}
	
	/**
	 * 根据已知hbase rowkey查询招商单位信息
	 * @param rowKey
	 * @return
	 */
	public HBGovernment getgovInfo(String rowKey){
		HBGovernment gov = null;
		Connection conn = null;
		Table table = null;
		try {
			conn = ConnectionFactory.createConnection(config);
			table = conn.getTable(this.tableNameMap.get(HbaseTableGovernment.TABLE_NAME));
			Get get = new Get(Bytes.toBytes(rowKey));
			Result result = table.get(get);
			if(result != null){
				gov = new HBGovernment();
				String[] ary = rowKey.split("_");
				gov.setProvincePostcode(ary[0]);
				gov.setCityPostcode(ary[1]);
				gov.setCountyPostcode(ary[2]);
				gov.setName(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY, HbaseTableGovernment.Column_Name_name));
				gov.setCity(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY, HbaseTableGovernment.Column_Name_city));
				gov.setAddress(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY, HbaseTableGovernment.Column_Name_address));
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			this.close(conn, table);
		}
		return gov;
	}
	
	/**
	 * 查询招商单位联系人
	 * @param provincePostcode
	 * @param cityPostcode
	 * @param countyPostcode
	 * @param govName
	 * @return
	 */
	public List<HBGovContacts> getContacts(String provincePostcode, String cityPostcode, String countyPostcode, String govName){
		List<HBGovContacts> contactList = new ArrayList<HBGovContacts>();
		Connection conn = null;
		Table table = null;
		try {
			conn = ConnectionFactory.createConnection(config);
			table = conn.getTable(this.tableNameMap.get(HbaseTableGovernment.TABLE_NAME));
			String startRow = provincePostcode + "_" + cityPostcode + "_" + countyPostcode + "_" + MD5Util.createMd5(govName) + "_0";
			String stopRow = provincePostcode + "_" + cityPostcode + "_" + countyPostcode + "_" + MD5Util.createMd5(govName) + "_z";
			Scan scan = new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow));
			ResultScanner rs = table.getScanner(scan);
			for(Result result: rs.next(100)){
				HBGovContacts contact = new HBGovContacts();
				contact.setProvincePostcode(provincePostcode);
				contact.setCityPostcode(cityPostcode);
				contact.setCountyPostcode(countyPostcode);
				contact.setName(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY_contact, HbaseTableGovernment.Column_Name_name));
				contact.setTelephone(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY_contact, HbaseTableGovernment.Column_Name_contact_telephone));
				contact.setGov(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY_contact, HbaseTableGovernment.Column_Name_contact_gov));
				contact.setDep(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY_contact, HbaseTableGovernment.Column_Name_contact_dep));
				contact.setFax(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY_contact, HbaseTableGovernment.Column_Name_contact_fax));
				contact.setEmail(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY_contact, HbaseTableGovernment.Column_Name_contact_email));
				contact.setQq(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY_contact, HbaseTableGovernment.Column_Name_contact_qq));
				contactList.add(contact);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			this.close(conn, table);
		}
		return contactList;
	}
	
	/**
	 * 根据已知的hbase rowkey查询联系人
	 * @param rowKey
	 * @return
	 */
	public HBGovContacts getContacts(String rowKey){
		HBGovContacts contact = null;
		Connection conn = null;
		Table table = null;
		try {
			conn = ConnectionFactory.createConnection(config);
			table = conn.getTable(this.tableNameMap.get(HbaseTableGovernment.TABLE_NAME));
			Get get = new Get(Bytes.toBytes(rowKey));
			Result result = table.get(get);
			if(result != null){
				contact = new HBGovContacts();
				String[] ary = rowKey.split("_");
				contact.setProvincePostcode(ary[0]);
				contact.setCityPostcode(ary[1]);
				contact.setCountyPostcode(ary[2]);
				contact.setName(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY_contact, HbaseTableGovernment.Column_Name_name));
				contact.setTelephone(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY_contact, HbaseTableGovernment.Column_Name_contact_telephone));
				contact.setGov(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY_contact, HbaseTableGovernment.Column_Name_contact_gov));
				contact.setDep(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY_contact, HbaseTableGovernment.Column_Name_contact_dep));
				contact.setFax(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY_contact, HbaseTableGovernment.Column_Name_contact_fax));
				contact.setEmail(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY_contact, HbaseTableGovernment.Column_Name_contact_email));
				contact.setQq(this.getRow(result, HbaseTableGovernment.COLUMN_FAMILY_contact, HbaseTableGovernment.Column_Name_contact_qq));
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			this.close(conn, table);
		}
		return contact;
	}
	
	/**
	 * 插入招商单位信息
	 * @param gov
	 * @return status
	 */
	public HbaseStatus insertGov(HBGovernment gov){
		HbaseStatus status = HbaseStatus.successed;
		Connection hconn = null;
		Table table = null;
		if(gov.getProvincePostcode() == null || gov.getProvincePostcode().length() == 0){
			return HbaseStatus.illogic;
		}
		try {
			if(gov.getCityPostcode() == null){
				gov.setCityPostcode("0");
			}
			if(gov.getCountyPostcode() == null){
				gov.setCountyPostcode("0");
			}
			hconn = ConnectionFactory.createConnection(config);
			table = hconn.getTable(this.tableNameMap.get(HbaseTableGovernment.TABLE_NAME));
			String rowKey = gov.getProvincePostcode() + "_" + gov.getCityPostcode() + "_" + gov.getCountyPostcode() + "_" + MD5Util.createMd5(gov.getName());
			Put put = new Put(Bytes.toBytes(rowKey));
			byte[] famByt = Bytes.toBytes(HbaseTableGovernment.COLUMN_FAMILY);
			this.addValue(put, famByt, HbaseTableGovernment.Column_Name_name, gov.getName());
			this.addValue(put, famByt, HbaseTableGovernment.Column_Name_city, gov.getCity());
			this.addValue(put, famByt, HbaseTableGovernment.Column_Name_address, gov.getAddress());
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
			status = HbaseStatus.exception;
		} finally{
			this.close(hconn, table);
		}
		return status;
	}
	
	/**
	 * 插入招商单位联系人
	 * @param contact
	 * @return status
	 */
	public HbaseStatus insertContact(HBGovContacts contact){
		HbaseStatus status = HbaseStatus.successed;
		Connection hconn = null;
		Table table = null;
		if((contact.getProvincePostcode() == null || contact.getProvincePostcode().length() == 0 
				 || contact.getGov() == null || contact.getGov().length() == 0)
				|| contact.getName() == null || contact.getName().length() == 0){
			return HbaseStatus.illogic;
		}
		try {
			if(contact.getCityPostcode() == null){
				contact.setCityPostcode("0");
			}
			if(contact.getCountyPostcode() == null){
				contact.setCountyPostcode("0");
			}
			hconn = ConnectionFactory.createConnection(config);
			table = hconn.getTable(this.tableNameMap.get(HbaseTableGovernment.TABLE_NAME));
			String rowKey;
			if(contact.getGov() != null && contact.getGov().matches("[0-1_]+")){
				rowKey = contact.getGov() +  "_" + MD5Util.createMd5(contact.getName());
			} else{
				rowKey = contact.getProvincePostcode() + "_" + contact.getCityPostcode() + "_" + contact.getCountyPostcode() + "_" + MD5Util.createMd5(contact.getGovName()) + "_" + MD5Util.createMd5(contact.getName());
			}
			Put put = new Put(Bytes.toBytes(rowKey));
			byte[] famByt = Bytes.toBytes(HbaseTableGovernment.COLUMN_FAMILY_contact);
			this.addValue(put, famByt, HbaseTableGovernment.Column_Name_contact_name, contact.getName());
			this.addValue(put, famByt, HbaseTableGovernment.Column_Name_contact_telephone, contact.getTelephone());
			this.addValue(put, famByt, HbaseTableGovernment.Column_Name_contact_gov, contact.getGov());
			this.addValue(put, famByt, HbaseTableGovernment.Column_Name_contact_govName, contact.getGovName());
			this.addValue(put, famByt, HbaseTableGovernment.Column_Name_contact_dep, contact.getDep());
			this.addValue(put, famByt, HbaseTableGovernment.Column_Name_contact_fax, contact.getFax());
			this.addValue(put, famByt, HbaseTableGovernment.Column_Name_contact_email, contact.getEmail());
			this.addValue(put, famByt, HbaseTableGovernment.Column_Name_contact_qq, contact.getQq());
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
			status = HbaseStatus.exception;
		} finally{
			this.close(hconn, table);
		}
		return status;
	}
	
	/**
	 * 赋值
	 * @param put
	 * @param famByt
	 * @param qualifier
	 * @param value
	 */
	private void addValue(Put put, byte[] famByt, String qualifier, String value){
		if(value != null){
			put.addColumn(famByt, Bytes.toBytes(qualifier), Bytes.toBytes(value));
		}
	}
	
	/**
	 * 获取指定列的值
	 * @param result
	 * @param familly
	 * @param rowName
	 * @return
	 */
	protected String getRow(Result result,String familly,String rowName){
		byte[] byts = result.getValue(Bytes.toBytes(familly), Bytes.toBytes(rowName));
		String row = null;
		if (byts != null && byts.length > 0) {
			row = Bytes.toString(byts);
		}
		return row;
	}

	protected void close(Connection hcon, Table table) {
		try {
			table.close();
		} catch (Exception e1) {		}
		try {
			hcon.close();
		} catch (Exception e) {		}
	}
}
